<?php
declare (strict_types = 1);
namespace app\common\library;

use PhpAmqpLib\Connection\AMQPStreamConnection;
use PhpAmqpLib\Message\AMQPMessage;
use PhpAmqpLib\Wire\AMQPTable;
use think\facade\Config;

class AMQP
{
    //MQ的默认连接配置
    public $config = [];

    public $connection; //链接
    public $channel; //信道

    public $exchangeName = ''; //交换机名
    public $queueName    = ''; //队列名
    public $routeKey     = ''; //路由键
    public $exchangeType = 'direct'; //交换机类型

    public $autoAck   = true; //是否自动ack应答
    private $callback = null;

    public function __construct($queue, $config = [])
    {
        if (is_string($queue)) {
            $configList = Config::get('eduline.queue');
            $queue      = $configList[$queue] ?? [];
        }
        if (!is_array($queue) || empty($queue)) {
            throw new \Exception("queue config is empty.");
        }

        $this->exchangeName = $queue['exchange_name'] ?? '';
        $this->exchangeType = $queue['type'] ?? 'direct';
        $this->queueName    = $queue['queue_name'] ?? '';
        $this->routeKey     = $queue['route_key'] ?? '';

        $defaultConfig = Config::get('eduline.rabbitmq', []);
        $config        = array_merge($defaultConfig, $config);
        if (empty($config)) {
            throw new \Exception("rabbitmq config is empty.");
        }

        $this->setConfig($config);

        $this->connect();
    }

    /**
     * 创建连接与信道
     * Author   Martinsun<syh@sunyonghong.com>
     * Date 2020-04-12
     * @return   [type]                         [description]
     */
    private function connect()
    {
        $host     = $this->config['host'];
        $port     = $this->config['port'];
        $user     = $this->config['user'];
        $password = $this->config['password'];
        $vhost    = $this->config['vhost'];
        if (empty($host) || empty($port) || empty($user) || empty($password)) {
            throw new \Exception('rabbitmq config is empty.');
        }
        //创建链接
        $this->connection = new AMQPStreamConnection($host, $port, $user, $password, $vhost, false, 'AMQPLAIN', null, 'en_US', 3.0, 3.0, null, false, 10);
        //创建信道
        $this->channel = $this->connection->channel();
        $this->createExchange();
        $this->createQueue();
    }

    /**
     * 创建交换机
     * Author   Martinsun<syh@sunyonghong.com>
     * Date 2020-04-12
     * @return   [type]                         [description]
     */
    private function createExchange()
    {
        $exchangeName = $this->exchangeName;
        $exchangeType = $this->exchangeType;
        if ($exchangeType == 'delayed') {
            // 延迟交换机
            $this->channel->exchange_declare(
                $exchangeName,
                // exchange类型为x-delayed-message
                'x-delayed-message',
                false,
                true,
                false,
                false,
                false,
                // 此处是重点，$argument必须使用new AMQPTable()生成
                new AMQPTable([
                    "x-delayed-type" => 'direct',
                ])
            );
        } else {
            // 常规交换机
            $this->channel->exchange_declare($exchangeName, $exchangeType, false, true, false);
        }
    }

    /**
     * 创建队列
     * Author   Martinsun<syh@sunyonghong.com>
     * Date 2020-04-13
     * @param    [type]                         $queueName [description]
     * @return   [type]                                    [description]
     */
    private function createQueue()
    {
        // passive: 消极处理， 判断是否存在队列，存在则返回，不存在直接抛出 PhpAmqpLib\Exception\AMQPProtocolChannelException 异常
        // durable：true、false true：在服务器重启时，能够存活
        // exclusive ：是否为当前连接的专用队列，在连接断开后，会自动删除该队列
        // autodelete：当没有任何消费者使用时，自动删除该队列
        // arguments: 自定义规则
        $this->channel->queue_declare($this->queueName, false, true, false, false);
        $this->channel->queue_bind($this->queueName, $this->exchangeName, $this->routeKey);
    }

    /**
     * 推送消息
     * Author   Martinsun<syh@sunyonghong.com>
     * Date 2020-04-12
     * @param    [type]                         $data [description]
     * @return   [type]                               [description]
     */
    public function push($handler, $data, $delayed = 0, $job_failed_count = 0)
    {
        if (!is_callable([$handler, 'run'])) {
            throw new \Exception('handler:[' . $handler . '] is not callable `run` function.');
        }

        $datas = [
            'handler'          => $handler,
            'data'             => $data,
            'job_failed_count' => $job_failed_count,
        ];

        if ($delayed > 0) {
            // 延迟消息
            $msg = new AMQPMessage(
                json_encode($datas),
                [
                    'delivery_mode'       => AMQPMessage::DELIVERY_MODE_PERSISTENT,
                    // 此处是重点，设置延时时间，单位是毫秒 1s=1000ms
                    'application_headers' => new AMQPTable([
                        'x-delay' => 1000 * $delayed,
                    ]),
                ]
            );
        } else {
            // 常规消息
            $msg = new AMQPMessage(
                json_encode($datas),
                [
                    'delivery_mode' => AMQPMessage::DELIVERY_MODE_PERSISTENT,
                ]
            );
        }

        $this->channel->basic_publish($msg, $this->exchangeName, $this->routeKey);
    }

    /**
     * 处理消息
     * Author   Martinsun<syh@sunyonghong.com>
     * Date 2020-04-12
     * @param    [type]                         $flag [description]
     * @return   [type]                               [description]
     */
    public function run(callable $callback)
    {
        $queueName = $this->queueName;
        $this->channel->basic_qos(null, 1, null);
        // 1:queue 要取得消息的队列名
        // 2:consumer_tag 消费者标签
        // 3:no_local false这个功能属于AMQP的标准,但是rabbitMQ并没有做实现.参考
        // 4:no_ack  false收到消息后,是否不需要回复确认即被认为被消费
        // 5:exclusive false排他消费者,即这个队列只能由一个消费者消费.适用于任务不允许进行并发处理的情况下.比如系统对接
        // 6:nowait false不返回执行结果,但是如果排他开启的话,则必须需要等待结果的,如果两个一起开就会报错
        // 7:callback  null回调函数
        // 8:ticket  null
        // 9:arguments null
        $this->channel->basic_consume($queueName, '', false, false, false, false, $callback);
        //监听消息
        while (count($this->channel->callbacks)) {
            $this->channel->wait();
        }
    }

    /**
     * 关闭连接
     * Author   Martinsun<syh@sunyonghong.com>
     * Date 2020-04-12
     * @return   [type]                         [description]
     */
    public function closeConnetct()
    {
        if ($this->connection) {
            $this->channel->close();
            $this->connection->close();
        }
    }

    /**
     * 重设配置
     * Author   Martinsun<syh@sunyonghong.com>
     * Date 2020-04-12
     * @param    array                          $config [description]
     */
    public function setConfig(array $config)
    {
        foreach ($config as $key => $value) {
            $this->config[$key] = $value;
        }
    }

    public function __destruct()
    {
        $this->closeConnetct();
    }
}
